package com.xiewei.kafka.chapter.chapter;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class KafkaProducer {
    private static final String KAFAKA_HOST = "127.0.0.1:9092";

    private org.apache.kafka.clients.producer.KafkaProducer producer;
    private static String topic = "MyTopic";


    public KafkaProducer() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,KAFAKA_HOST);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "producer");
        this.producer = new org.apache.kafka.clients.producer.KafkaProducer(props);
    }
    public void sendMsg(){

        producer.send(new ProducerRecord(topic, 1, "message"), new Callback() {
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                System.out.println("message send to :["+ recordMetadata.partition()+"],offset:["+recordMetadata.offset()+"]");
            }
        });
    }
    public static void main(String[] args) throws InterruptedException {
        KafkaProducer producer = new KafkaProducer();
        while (true) {
            Thread.sleep(1000);
            producer.sendMsg();
        }


    }
}
